/*
 * collector.h
 *
 *  Created on: Mar 21, 2012
 *      Author: yxjiang
 */

#ifndef COLLECTOR_H_
#define COLLECTOR_H_

#include <pthread.h>
#include <netdb.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sstream>
#include <list>
#include <map>
#include <boost/property_tree/json_parser.hpp>
#include <boost/property_tree/ptree.hpp>
#include <boost/shared_ptr.hpp>
#include "../common/common.h"

namespace event
{
/*
 * Use shared pointer to store the large property_tree object
 */
typedef boost::shared_ptr<boost::property_tree::ptree> SharedPtree;

/*
 * The data stream that grab the meta-data from remote target.
 */
class DataStream
{
public:
	/**
	 * Create a data stream to get data from a specified source.
	 * name		-	the name of the stream.
	 * ip		-	the IP of the source.
	 * port		-	the port of the source.
	 * time_range_in_sec	-	the time range that the stream buffered. The time unit is second.
	 */
	DataStream(const std::string &name, const std::string &ip, int port, int time_range_in_sec);
	~DataStream();
	/*
	 * Get the name of the stream.
	 */
	std::string GetName() const;
	/*
	 * Start to grab the data from monitor.
	 * Return:
	 * 		true if stream source is available, false if not.
	 */
	bool GrabData();
	/*
	 * Get a snapshot of the stream.
	 */
	boost::shared_ptr<std::list<SharedPtree> > GetStreamSnapshot();

private:
	std::string name_;
	std::list<SharedPtree> stream_;	//	head is old, tail is new
	static pthread_rwlock_t stream_lock_;
	std::string stream_ip_;
	int stream_port_;
	int time_range_in_sec_;
	struct sockaddr_in server_addr_;
};


/*
 * The basic data structure to record the status of stream.
 */
typedef struct
{
	DataStream *stream;	//	the object of stream
	bool running;		//	whether the stream is currently running
	pthread_t pid;		//	the pthread id of running thread of the stream
	int misses;	//	how many continuous times the stream is unavailable
} StreamStatus;


/*
 * The collector that manages the streams.
 */
class Collector
{
public:
	/*
	 * Set the rate of grab data.
	 * Rate is defined the interval of each grab in unit of seconds.
	 */
	Collector(int rate, int port, int buffer_time_len, int retrieve_service_port);

	static void SetRate(int rate);

	/*
	 * Start all service.
	 */
	void Start();

	/*
	 * Run each stream as a thread.
	 * Continuously notify the stream to grab.
	 */
	static void _StartStreams();

private:
	/*
	 * Service of registration.
	 */
	static void *_Registration(void *arg);
	/*
	 * Register the stream to the collector.
	 */
	static bool _RegisterStream(DataStream *stream);
	/*
	 * The work thread of the stream.
	 */
	static void *_StreamService(void *arg);
	/*
	 * Serve the retrieve request from the client.
	 */
	static void *_RetrieveService(void *arg);



private:
	static std::map<std::string, StreamStatus> streams_;
	static pthread_mutex_t streams_lock_;
	static int buffer_time_length_;		//	the time length for buffer
	static int rate_;
	int registration_port_;
	int retrieve_service_port_;
	pthread_t registration_pid_;	//	the pid of the registration service
	pthread_t retrieve_service_pid_;			//	the pid of the retrieve service
};

};

#endif /* COLLECTOR_H_ */
